﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace XcoAppSpaces.Core.Ports.Utils
{
	/// <summary>
	/// Version of the BufferBlock that is using a ConcurrentQueue.
	/// </summary>
	/// <typeparam name="T"></typeparam>
	class PortBufferCQ<T>
    {
		private readonly ConcurrentQueue<T> _items = new ConcurrentQueue<T>();
		private Queue<TaskCompletionSource<T>> _waitingReceivers; //is instantiated on demand

		private bool _registeredReceiver;
		private Action<T> _receiverAction;
		private Func<T, Task> _receiverFunc;
		private int _maxDegreeOfParallelism;
		private int _numberOfCurrentlyRunningProcessors;
		private readonly object _processingStarterLock = new object();

		private TaskScheduler _scheduler;
		
		public int Count { get { return _items.Count; } }

		public void Post(T item)
		{
			if (_registeredReceiver)
			{
				_items.Enqueue(item);
				StartProcessing();
			}
			else
			{
				lock (_items)
				{
					while (_waitingReceivers != null && _waitingReceivers.Count > 0)
					{
						var awaiter = _waitingReceivers.Dequeue();
						//if giving the item to an awaiter was successful, stop here
						//if not, try the next one (could be unsuccessful e.g. if the awaiter has been cancelled because of a timeout)
						if (awaiter.TrySetResult(item))
							return;
					}
					//if the item could not be given to any awaiter, enqueue it
					_items.Enqueue(item);
				}
			}
		}

		public Task<T> ReceiveAsync()
		{
			return ReceiveAsync(0);
		}

		public Task<T> ReceiveAsync(int timeout)
		{
			if (_registeredReceiver)
				throw new Exception("Cannot receive items with ReceiveAsync() while a receiver is registiered!");
			lock (_items)
			{
				T item;
				if (_items.TryDequeue(out item))
#if SILVERLIGHT
					return TaskEx.FromResult(item);
#else
					return Task.FromResult(item);
#endif

				var awaiter = new TaskCompletionSource<T>();
				if (_waitingReceivers == null)
					_waitingReceivers = new Queue<TaskCompletionSource<T>>();
				_waitingReceivers.Enqueue(awaiter);

				if (timeout > 0)
				{
					var ct = new CancellationTokenSource(timeout);
					ct.Token.Register(() => awaiter.TrySetException(new TimeoutException()), useSynchronizationContext: false);
				}

				return awaiter.Task;
			}
		}

		public void Clear()
		{
			lock (_items)
			{
				_items.Clear();
			}
		}

		public void RegisterReceiver(Action<T> blockAction, PortReceiverOptions options)
		{
			_receiverAction = blockAction;
			_receiverFunc = null;
			_maxDegreeOfParallelism = options.MaxDegreeOfParallelism;
			_scheduler = options.TaskScheduler;
			_registeredReceiver = true;
			if (_items.Count > 0)
				StartProcessing();
		}

		public void RegisterReceiver(Func<T, Task> blockFunc, PortReceiverOptions options)
		{
			_receiverFunc = blockFunc;
			_receiverAction = null;
			_maxDegreeOfParallelism = options.MaxDegreeOfParallelism;
			_scheduler = options.TaskScheduler;
			_registeredReceiver = true;
			if (_items.Count > 0)
				StartProcessing();
		}

		private void StartProcessing()
		{
			lock (_processingStarterLock)
			{
				if (_maxDegreeOfParallelism != 0 && _numberOfCurrentlyRunningProcessors >= _maxDegreeOfParallelism)
					return;
				_numberOfCurrentlyRunningProcessors++;
			}
			if (_receiverAction != null)
			{
				if (_scheduler != null)
					Task.Factory.StartNew((Action) ActionProcessor, new CancellationToken(), TaskCreationOptions.None, _scheduler);
				else
					Task.Factory.StartNew((Action) ActionProcessor);

			}
			else
			{
				if (_scheduler != null)
					Task.Factory.StartNew((Func<Task>)FuncProcessor, new CancellationToken(), TaskCreationOptions.None, _scheduler);
				else
					Task.Factory.StartNew((Func<Task>)FuncProcessor);
			}
		}

		private void ActionProcessor()
		{
			while (true)
			{
				T item;
				if (!_items.TryDequeue(out item))
				{
					lock (_processingStarterLock)
					{
						_numberOfCurrentlyRunningProcessors--;
					}
					return;
				}
				_receiverAction(item);
			}
		}

		private async Task FuncProcessor()
		{
			while (true)
			{
				T item;
				if (!_items.TryDequeue(out item))
				{
					lock (_processingStarterLock)
					{
						_numberOfCurrentlyRunningProcessors--;
					}
					return;
				}
				await _receiverFunc(item).ConfigureAwait(false);
			}
		}

		public void UnregisterReceiver()
		{
			_registeredReceiver = false;
			_receiverAction = null;
			_receiverFunc = null;
		}
    }
}
